package com.ruoyi.eqmonitor;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.json.JSONUtil;
import cn.hutool.poi.excel.ExcelUtil;
import cn.hutool.poi.excel.ExcelWriter;
import com.ruoyi.eqmonitor.dispose.domain.OpentsdbTags;
import com.ruoyi.eqmonitor.dtos.OptDeviceAlarmCountDto;
import com.ruoyi.eqmonitor.dtos.OptDeviceDto;
import com.ruoyi.eqmonitor.service.IOptDeviceService;
import com.ruoyi.eqmonitor.utils.Constant;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.nio.reactor.IOReactorException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.opentsdb.client.OpenTSDBClient;
import org.opentsdb.client.bean.request.Query;
import org.opentsdb.client.bean.request.SubQuery;
import org.opentsdb.client.bean.response.QueryResult;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * Opentsdb Demo
 * 注意 ** opentsdb查询, 如果此设备没有任何指标被存储, 会抛出异常, 所以需要在所有使用Opentsdb的位置做try catch(调用OpenTSDB http api发生错误，响应码:400,错误信息:No such name for 'tagv': '0x0502') **
 */
@Slf4j
@SpringBootTest
@RunWith(SpringRunner.class)
public class OpenTSDBDemo {

    @Resource
    private OpenTSDBUtils openTSDBUtils;
    @Resource
    private IOptDeviceService iOptDeviceService;

    /**
     * opentsdb查询client
     */
    private OpenTSDBClient queryClient;

    String startTime = "2022-11-20 00:00:00";
    String endTime = "2022-11-20 23:59:59";

    /** 测试设备编码 */
    String[] snCodes = {
//            // 两个液位计
//            "3M0WQ6", "NAT837",
            // 原位水质监测仪（COD+液位）
            "P5D3N5",
            // 流量水质（氨氮）一体机
            "U4D2E8",
            // 管道流量监测仪
            "0A9BK7",
    };

    /** 测试监测指标 */
    String[] monitorCodes = {
//            // 液位查询 (液位高度)
//            "0x05D1",
//            // 水质查询 (温度, 湿度, PH, ORP, DO, 氨氮)
//            "0x0502", "0x0506", "0x0522", "0x0524","0x0526", "0x0528",
            // 流量查询 (设备实时电压, 温度, 瞬时流量, 流速, 液位高度)
            /*"0x0501", "0x0502", "0x0592", "0x0593",*/ "0x05D1"
    };

    @Before
    public void init() throws IOReactorException {
        try {
            queryClient = openTSDBUtils.readClient();
        } catch (IOReactorException e) {
            log.error("创建opentsdb查询失败");
            throw e;
        }
    }

    @After
    public void close() {
        if (queryClient != null) {
            //关闭连接
            try {
                queryClient.gracefulClose();
            } catch (IOException e) {
                log.error("关闭opentsdb连接失败");
            }
        }
    }

    /**
     * 查询某个设备的所有最新监测指标 (需要自行在dps中过滤最后一个)
     * ** 适用于大屏类, 比如获取某一个或者某几个设备的所有指标最后一次传递的值 **
     * 返回值示例:
     * [
     *     {
     *         "dps":{
     *             "1668307393":0,
     *             "1668315495":0,
     *             "1668314595":0,
     *             "1668319995":0,
     *             "1668316395":0,
     *             "1668310093":0,
     *             "1668309193":0,
     *             "1668319095":0,
     *             "1668320899":0,
     *             "1668308293":0
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"瞬时流量",
     *             "dataUnit":"m3/s",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0592"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668314595":0.227,
     *             "1668319995":0.263,
     *             "1668318195":0.252,
     *             "1668310093":0.195,
     *             "1668319095":0.258,
     *             "1668320899":0.269,
     *             "1668308293":0.182
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"液位高度",
     *             "dataUnit":"m",
     *             "equipmentType":"0",
     *             "monitorCode":"0x05D1"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668307393":0,
     *             "1668315495":0,
     *             "1668314595":0,
     *             "1668319995":0,
     *             "1668317295":0,
     *             "1668316395":0,
     *             "1668310093":0,
     *             "1668309193":0,
     *             "1668319095":0,
     *             "1668320899":0,
     *             "1668308293":0
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"流速",
     *             "dataUnit":"m/s",
     *             "equipmentType":"0",
     *             "monitorCode":"0x0593"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668320899":13.72
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"设备实时电压",
     *             "dataUnit":"V",
     *             "equipmentType":"0",
     *             "monitorCode":"0x0501"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     }
     * ]
     */
    @Test
    public void queryLastMonitorData() {
        // 1. 创建snCodeFilter和monitorCode过滤器
        String[] codes = {"6SQK78",
                "GG8N76",
                "67WW0R",
                "K61P0F",
                "LP9F04",
                "M17WP3",
                "PA674U"};
//        String[] mCodes = {"8A10RB", "L14LL4", "79HB5U", "KT7G26", "504QAP", "2DU38C", "YA4A53", "3P4EJ3", "ESZ812", "UP8R64"};
        SubQuery.Filter snCodeFilter= OpenTSDBUtils.createSnCodeFilter(codes);
//        SubQuery.Filter monitorCodeFilter= createMonitorCodeFilter("0x0566");
        List<Map<String, Object>> excelList = new ArrayList<>();
        // 2. 生成查询, 默认查询2天内最近的数据
        Query query = Query
                // begin可以写时间戳(秒格式)或固定的几个,请参见readme中的参考文档链接
                .begin("1d-ago")
                // 必须使用时区日历
                .useCalendar()
                // 必须使用数据存入时的时区
                .timezone(OpenTSDBUtils.OPENTSDB_TIMEZONE)
                // 子查询
                .sub(SubQuery
                        // metric 类似于sql中的数据库的含义
                        .metric(openTSDBUtils.getMetric())
                        .aggregator(SubQuery.Aggregator.NONE)
                        .filter(snCodeFilter)
//                        .downsample("24h-max-zero")
//                        .filter(monitorCodeFilter)
                        .build()
                ).build();
        try {
            // 3. 查询
            List<QueryResult> resultList = queryClient.query(query);
            for (QueryResult queryResult : resultList) {
                // 目的是将tags的base64转换为原始的字符串
                OpentsdbTags opentsdbTags = OpenTSDBUtils.decodeToBean(queryResult.getTags());
//                log.info("最新监测指标查询结果: tags: {}\n dps: {}", opentsdbTags, queryResult.getDps());
                queryResult.getDps().forEach((stamp, number) -> {
                    Map<String, Object> map = new HashMap<>();
                    map.put("设备编码", opentsdbTags.getSnCode());
                    map.put("设备类型", opentsdbTags.getEquipmentTypeCode());
                    map.put("设备类型名称", opentsdbTags.getEquipmentType());
                    map.put("监测指标", opentsdbTags.getMonitorCode());
                    map.put("监测指标名称", opentsdbTags.getMonitorName());
                    map.put("时间(秒级时间戳)", stamp);
                    map.put("时间(时间)", OpenTSDBUtils.deserialize(stamp));
                    map.put("数据值", number);
                    map.put("单位", opentsdbTags.getDataUnit());
                    excelList.add(map);
                });
            }
            ExcelWriter writer = ExcelUtil.getWriter("D:\\desktop\\7台河道监测站_流量_液位_24小时.xlsx");
            ExcelWriter write = writer.write(excelList, true);
            System.out.println(excelList.size());
            write.flush();
            write.close();
//            log.info("opentsdb查询结果: {}", JSONUtil.toJsonStr(resultList));
        } catch (IOException | ExecutionException | InterruptedException e) {
            log.error("查询opentsdb失败", e);
        }
    }

    /**
     * 降精度聚合采样
     * ** 适用于分析类, 比如分析某个或某几个液位计, 本月的平均液位高度等 **
     * 返回值示例:
     * [
     *     {
     *         "dps":{
     *             "1668268800":13.72
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"设备实时电压",
     *             "dataUnit":"V",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0501"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668268800":13.78
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"1JPX24",
     *             "monitorName":"设备实时电压",
     *             "dataUnit":"V",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0501"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     }
     * ]
     */
    @Test
    public void downsampleQuery() {
        // 降精度的类型, 具体参考readme中的链接资料
        String downsample = "1d-avg-zero";

        // 1. 创建snCodeFilter和monitorCode过滤器
        SubQuery.Filter snCodeFilter = OpenTSDBUtils.createSnCodeFilter(snCodes);
        SubQuery.Filter monitorCodeFilter = OpenTSDBUtils.createMonitorCodeFilter(monitorCodes);

        // 2. 计算时间
        long startDateTime = OpenTSDBUtils.serializedTime(startTime);
        long endDateTime = OpenTSDBUtils.serializedTime(endTime);
        // 如果结束时间比当前时间还大, 结束时间则取当前时间(/1000的原因之前有介绍过, opentsdb查询时间戳是以秒为单位的时间戳)
        long now = new Date().getTime() / 1000;
        if (endDateTime > now){
            endDateTime = now;
        }

        // 3. 子查询
        SubQuery sub = SubQuery.metric(openTSDBUtils.getMetric())
                // avg平均值聚合
                .aggregator(SubQuery.Aggregator.NONE)
                // 使用声明的downsample进行采样
                .downsample(downsample)
                // 传入设备编码和监测指标编码的过滤条件
                .filter(snCodeFilter)
                .filter(monitorCodeFilter)
                .build();

        // 4. 查询主体
        Query queryParams = Query
                // 开始时间 秒
                .begin(startDateTime)
                // 结束时间 秒
                .end(endDateTime)
                // 时区相关
                .useCalendar().timezone(OpenTSDBUtils.OPENTSDB_TIMEZONE)
                // 子查询
                .sub(sub)
                .build();

        try {
            // 5. 进行查询
            List<QueryResult> resultList = queryClient.query(queryParams);
            for (QueryResult queryResult : resultList) {
                // 目的是将tags的base64转换为原始的字符串
                OpentsdbTags opentsdbTags = OpenTSDBUtils.decodeToBean(queryResult.getTags());
                log.info("降精度采样查询结果: tags: {}\n dps: {}", opentsdbTags, queryResult.getDps());
            }
//            log.info("降精度采样查询结果: {}", JSONUtil.toJsonStr(resultList));
        } catch (IOException | ExecutionException | InterruptedException e) {
            log.error("查询opentsdb失败", e);
        }
    }

    /**
     * 查询时序列表
     * 与降精度采样基本一致, 不同内容为降精度采样需要聚合, 列表不需要聚合
     * ** 适用于列表类, 比如查询本天某个或某几个设备 的 某一个或某几个监测指标的时序数据列表 **
     * 返回值示例:
     * [
     *     {
     *         "dps":{
     *             "1668307393":0,
     *             "1668315495":0,
     *             "1668314595":0,
     *             "1668319995":0,
     *             "1668316395":0,
     *             "1668310093":0,
     *             "1668309193":0,
     *             "1668319095":0,
     *             "1668320899":0,
     *             "1668308293":0
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"瞬时流量",
     *             "dataUnit":"m3/s",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0592"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668314595":0.227,
     *             "1668319995":0.263,
     *             "1668318195":0.252,
     *             "1668310093":0.195,
     *             "1668319095":0.258,
     *             "1668320899":0.269,
     *             "1668308293":0.182
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"瞬时流量",
     *             "dataUnit":"m",
     *             "equipmentType":"0",
     *             "monitorCode":"0x05D1"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668307393":0,
     *             "1668315495":0,
     *             "1668314595":0,
     *             "1668319995":0,
     *             "1668317295":0,
     *             "1668316395":0,
     *             "1668310093":0,
     *             "1668309193":0,
     *             "1668319095":0,
     *             "1668320899":0,
     *             "1668308293":0
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"流速",
     *             "dataUnit":"m/s",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0593"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668320899":13.72
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"0C21JD",
     *             "monitorName":"设备实时电压",
     *             "dataUnit":"V",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0501"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668315519":0,
     *             "1668314619":0,
     *             "1668316419":0,
     *             "1668318219":0,
     *             "1668319119":0,
     *             "1668320023":0
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"1JPX24",
     *             "monitorName":"瞬时流量",
     *             "dataUnit":"m3/s",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0592"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668315519":1.13,
     *             "1668314619":1.124,
     *             "1668317319":0,
     *             "1668316419":0,
     *             "1668318219":0,
     *             "1668319119":1.146,
     *             "1668320023":1.15
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"1JPX24",
     *             "monitorName":"瞬时流量",
     *             "dataUnit":"m",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x05D1"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668314619":0,
     *             "1668317319":0,
     *             "1668316419":0,
     *             "1668319119":0,
     *             "1668313719":0
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"1JPX24",
     *             "monitorName":"流速",
     *             "dataUnit":"m/s",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0593"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     },
     *     {
     *         "dps":{
     *             "1668320023":13.78
     *         },
     *         "aggregateTags":[],
     *         "tags":{
     *             "snCode":"1JPX24",
     *             "monitorName":"设备实时电压",
     *             "dataUnit":"V",
     *             "equipmentType":"流量计",
     *             "monitorCode":"0x0501"
     *         },
     *         "metric":"monitor.equipment_tzsw"
     *     }
     * ]
     */
    @Test
    public void listQuery() {
        // 1. 创建snCodeFilter和monitorCode过滤器
        SubQuery.Filter snCodeFilter = OpenTSDBUtils.createSnCodeFilter(snCodes);
        SubQuery.Filter monitorCodeFilter = OpenTSDBUtils.createMonitorCodeFilter(monitorCodes);

        // 2. 计算时间
        long startDateTime = OpenTSDBUtils.serializedTime(startTime);
        long endDateTime = OpenTSDBUtils.serializedTime(endTime);
        // 如果结束时间比当前时间还大, 结束时间则取当前时间(/1000的原因之前有介绍过, opentsdb查询时间戳是以秒为单位的时间戳)
        long now = new Date().getTime() / 1000;
        if (endDateTime > now){
            endDateTime = now;
        }

        // 3. 子查询
        SubQuery sub = SubQuery
                .metric(openTSDBUtils.getMetric())
                // 不聚合
                .aggregator(SubQuery.Aggregator.NONE)
                .filter(snCodeFilter)
                .filter(monitorCodeFilter)
                .build();

        // 4. 查询主体
        Query queryParams = Query
                // 开始时间 秒
                .begin(startDateTime)
                // 结束时间 秒
                .end(endDateTime)
                // 时区相关
                .useCalendar().timezone(OpenTSDBUtils.OPENTSDB_TIMEZONE)
                // 子查询
                .sub(sub)
                .build();

        try {
            // 5. 进行查询
            List<QueryResult> resultList = queryClient.query(queryParams);
            for (QueryResult queryResult : resultList) {
                // 目的是将tags的base64转换为原始的字符串
                OpentsdbTags opentsdbTags = OpenTSDBUtils.decodeToBean(queryResult.getTags());
                log.info("列表查询结果: tags: {}\n dps: {}", opentsdbTags, queryResult.getDps());
            }
//            log.info("列表查询结果: {}", JSONUtil.toJsonStr(resultList));
        } catch (IOException | ExecutionException | InterruptedException e) {
            log.error("查询opentsdb失败", e);
        }
    }


    /**
     * 查询云臻需要的流量 示例
     */
    @Test
    public void testSelectFlow() {
        // 查询泵站周边的流量监测仪
        List<OptDeviceDto> bzFlowDevice = iOptDeviceService.getBzFlowDevice();
        // 过滤所有设备编码
        String[] deviceCodes = bzFlowDevice.stream().map(OptDeviceDto::getDeviceCode).distinct().toArray(String[]::new);
        // 1. 创建snCodeFilter过滤器
        SubQuery.Filter snCodeFilter = OpenTSDBUtils.createSnCodeFilter(deviceCodes);
        // 2. 创建monitorCode过滤器 过滤瞬时流量和正向累计流量
        SubQuery.Filter monitorCodeFilter = OpenTSDBUtils.createMonitorCodeFilter(Constant.MonitorInfo.M_0x0592.getCode(), Constant.MonitorInfo.M_0x0590.getCode());

        // 3. 子查询
        SubQuery sub = SubQuery
                .metric(openTSDBUtils.getMetric())
                // 不聚合
                .aggregator(SubQuery.Aggregator.NONE)
                .filter(snCodeFilter)
                .filter(monitorCodeFilter)
                .build();

        // 4. 查询主体
        Query queryParams = Query
                // 查询最近24小时
                .begin("24h-ago")
                // 时区相关
                .useCalendar().timezone(OpenTSDBUtils.OPENTSDB_TIMEZONE)
                // 子查询
                .sub(sub)
                .build();

        try {
            // 5. 进行查询
            List<QueryResult> resultList = queryClient.query(queryParams);
            for (QueryResult queryResult : resultList) {
                // 目的是将tags的base64转换为原始的字符串
                OpentsdbTags opentsdbTags = OpenTSDBUtils.decodeToBean(queryResult.getTags());
                log.info("过滤瞬时流量和正向累计流量列表查询结果: tags: {}\n dps: {}", opentsdbTags, queryResult.getDps());
            }
//            log.info("列表查询结果: {}", JSONUtil.toJsonStr(resultList));
        } catch (IOException | ExecutionException | InterruptedException e) {
            log.error("查询opentsdb失败", e);
        }
    }


    /**
     * 云臻需要的 10台氨氮+流量 是断面的设备  示例
     */
    @Test
    public void testSelectAnDanLiuLiang() {
        // 10台氨氮+流量设备
        OptDeviceDto optDeviceDto = new OptDeviceDto();
        optDeviceDto.setDeviceType(Constant.EquipmentType.SR500_L_Y0_LDN.getCode());
        List<OptDeviceDto> llAnDanFlowDevice = iOptDeviceService.selectOptDeviceList(optDeviceDto);
        // 过滤所有设备编码
        String[] deviceCodes = llAnDanFlowDevice.stream().map(OptDeviceDto::getDeviceCode).distinct().toArray(String[]::new);
        // 1. 创建snCodeFilter过滤器
        SubQuery.Filter snCodeFilter = OpenTSDBUtils.createSnCodeFilter(deviceCodes);
        // 2. 创建monitorCode过滤器 过滤氨氮和液位高度
        SubQuery.Filter monitorCodeFilter = OpenTSDBUtils.createMonitorCodeFilter(Constant.MonitorInfo.M_0x0528.getCode(), Constant.MonitorInfo.M_0x05D1.getCode());

        // 3. 子查询
        SubQuery sub = SubQuery
                .metric(openTSDBUtils.getMetric())
                // 不聚合
                .aggregator(SubQuery.Aggregator.NONE)
                .filter(snCodeFilter)
                .filter(monitorCodeFilter)
                .build();

        // 4. 查询主体
        Query queryParams = Query
                // 查询最近24小时
                .begin("24h-ago")
                // 时区相关
                .useCalendar().timezone(OpenTSDBUtils.OPENTSDB_TIMEZONE)
                // 子查询
                .sub(sub)
                .build();

        try {
            // 5. 进行查询
            List<QueryResult> resultList = queryClient.query(queryParams);
            for (QueryResult queryResult : resultList) {
                // 目的是将tags的base64转换为原始的字符串
                OpentsdbTags opentsdbTags = OpenTSDBUtils.decodeToBean(queryResult.getTags());
                log.info("过滤氨氮和液位高度列表查询结果: tags: {}\n dps: {}", opentsdbTags, queryResult.getDps());
            }
//            log.info("列表查询结果: {}", JSONUtil.toJsonStr(resultList));
        } catch (IOException | ExecutionException | InterruptedException e) {
            log.error("查询opentsdb失败", e);
        }
    }
}
